#pragma once

#include "mqttClient.hpp"
#include "mqttMsg.hpp"
#include "msgFactory.hpp"
#include "threadPool.hpp"
#include <condition_variable>
#include <iostream>
#include <list>
#include <memory>
#include <mutex>
#include <thread>

class MqttServer {
private:
    std::shared_ptr<MqttClient> _mqttClient;
    std::shared_ptr<DeviceManager> _deviceManager;
    std::list<MqttMsg> _queue;
    std::mutex _mutex;
    std::condition_variable _cv;
    std::unique_ptr<ThreadPool> _pool;

public:
    MqttServer(std::shared_ptr<MqttClient> m, std::shared_ptr<DeviceManager> d)
        : _mqttClient(m),
          _deviceManager(d),
          _pool(std::make_unique<ThreadPool>(10)) {
        _mqttClient->registerOnConnect([this](const auto &clientId) {
            std::cout << clientId << " connect success!!!" << std::endl;
            _mqttClient->subscribe("v1/dev/+/dev_dlt698/dev/add");
            _mqttClient->subscribe("v1/dev/+/+/point/get");
            _mqttClient->subscribe("v1/dev/+/+/point/set");
            _mqttClient->subscribe("v1/dev/sdev_manager/dev_dlt698/list/get_ack");
            _mqttClient->subscribe("v1/dev/sdev_manager/0/list/notify");
            _mqttClient->subscribe("v1/dev/+/+/state/get");
        });
        _mqttClient->registerOnPublish([](const auto &clientId, const auto &topic, const auto &payload) {
            std::cout << clientId << " publish topic --> " << topic << std::endl;
            std::cout << clientId << " publish payload --> " << payload << std::endl;
        });
        _mqttClient->registerOnSubscribe(
            [](const auto &clientId, const auto &topic) { std::cout << clientId << " subscribe " << topic << " success!!!" << std::endl; });
        _mqttClient->registerOnDisconnect([](const auto &clientId) { std::cout << clientId << " disconnect!!!" << std::endl; });
        _mqttClient->registerOnMessage([this](const auto &clientId, const auto &topic, const auto &payload) {
            {
                std::lock_guard<std::mutex> _lock(_mutex);
                _queue.emplace_back(topic, payload);
                _cv.notify_one();
            }
        });

        _mqttClient->connect("localhost");
    }

    void start(void) {
        std::thread([&] {
            while (true) {
                MqttMsg msg;
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    _cv.wait(_lock, [&] { return !_queue.empty(); });
                    msg = _queue.front();
                    _queue.pop_front();
                }

                _pool->submit(
                    [this](const MqttMsg &m) {
                        std::cout << "recv topic --> " << m.topic() << std::endl;
                        std::cout << "recv payload --> " << m.payload() << std::endl;
                        auto msgObj = Protocol::MsgFactory::create(m.topic(), m.payload(), _mqttClient, _deviceManager);
                        if (msgObj->parse(m.topic(), m.payload())) {
                            msgObj->action().reply();
                        }
                    },
                    msg);
            }
        }).detach();
    }
};
